package com.atguigu.app.dwd.db;

import com.atguigu.util.MyKafkaUtil_lj;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author 晶晶
 * @Date 2023 03 27 19 19
 **/
public class Dwd_Trade_OrderDetail {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("" +
                "create table topic_db(\n" +
                "   `database` string,\n" +
                "   `table` string,\n" +
                "   `type` string,\n" +
                "   `data` map<string,string>,\n" +
                "   `old` map<string,string>,\n" +
                "   `ts` string,\n" +
                "   `proc_time` as PROCTIME()\n" +
                ")"+ MyKafkaUtil_lj.getKafkaDDL("topic_db","trade_order_detail"));

        Table orderDetailTable = tableEnv.sqlQuery("" +
                "select\n" +
                "   `data`['id'] id,\n" +
                "   `data`['course_id'] course_id,\n" +
                "   `data`['course_name'] course_name,\n" +
                "   `data`['order_id'] order_id,\n" +
                "   `data`['user_id'] user_id,\n" +
                "   `data`['origin_amount'] origin_amount,\n" +
                "   `data`['coupon_reduce'] coupon_reduce,\n" +
                "   `data`['final_amount'] final_amount,\n" +
                "   `data`['create_time'] create_time,\n" +
                "   `data`['update_time'] update_time\n" +
                "from topic_db \n" +
                "where `database`='edu'\n" +
                "and `table`='order_detail'\n" +
                "and `type`='insert'");

        tableEnv.createTemporaryView("order_detail",orderDetailTable);

        Table orderInfoTable = tableEnv.sqlQuery("" +
                "select\n" +
                "   `data`['id'] id,\n" +
                "   `data`['user_id'] user_id,\n" +
                "   `data`['origin_amount'] origin_amount,\n" +
                "   `data`['coupon_reduce'] coupon_reduce,\n" +
                "   `data`['final_amount'] final_amount,\n" +
                "   `data`['order_status'] order_status,\n" +
                "   `data`['session_id'] session_id,\n" +
                "   `data`['province_id'] province_id\n" +
                "from topic_db\n" +
                "where `database`='edu'\n" +
                "and `table`='order_info'\n" +
                "and `type`='insert'");

        tableEnv.createTemporaryView("order_info",orderInfoTable);

        Table resultTable = tableEnv.sqlQuery("" +
                "select\n" +
                "   od.id,\n" +
                "   od.course_id,\n" +
                "   od.course_name,\n" +
                "   od.order_id,\n" +
                "   od.user_id,\n" +
                "   oi.session_id,\n" +
                "   oi.province_id,\n" +
                "   od.origin_amount,\n" +
                "   od.coupon_reduce,\n" +
                "   od.final_amount,\n" +
                "   od.create_time,\n" +
                "   od.update_time,\n" +
                "   oi.order_status\n" +
                "from order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id");

        tableEnv.createTemporaryView("result_table",resultTable);

        tableEnv.executeSql("" +
                "create table dwd_trade_order_detail (\n" +
                "   `id` string,\n" +
                "   `course_id` string,\n" +
                "   `course_name` string,\n" +
                "   `order_id` string,\n" +
                "   `user_id` string,\n" +
                "   `session_id` string,\n" +
                "   `province_id` string,\n" +
                "   `origin_amount` string,\n" +
                "   `coupon_reduce` string,\n" +
                "   `final_amount` string,\n" +
                "   `create_time` string,\n" +
                "   `update_time` string,\n" +
                "   `order_status` string\n" +
                ")"+MyKafkaUtil_lj.getKafkaSinkDDL("dwd_trade_order_detail_aa"));

        tableEnv.executeSql("insert into dwd_trade_order_detail select * from result_table");
    }
}
